from mrjob.job import MRJob


class ProductClick(MRJob):
    def mapper(self, _, line):
        lineinfo = line.strip().split(',')
        if len(lineinfo) == 5:
            if lineinfo[3] == 'pv':
                yield lineinfo[1], 1

    def combiner(self, pv, counts):
        yield pv, sum(counts)

    def reducer(self, pv, counts):
        yield pv, sum(counts)


if __name__ == '__main__':
    ProductClick.run()
